package com.dht.xiao.jianye.base.mqtt.core;

import android.content.Context;
import android.text.TextUtils;
import android.util.ArrayMap;

import org.eclipse.paho.android.service.MqttAndroidClient;
import org.eclipse.paho.client.mqttv3.IMqttActionListener;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.IMqttMessageListener;
import org.eclipse.paho.client.mqttv3.IMqttToken;
import org.eclipse.paho.client.mqttv3.MqttCallbackExtended;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;

import java.util.Iterator;
import java.util.Map;
import java.util.concurrent.TimeUnit;

import com.dht.xiao.jianye.base.exception.ApiException;
import com.dht.xiao.jianye.base.exception.ApiExceptionCode;
import com.dht.xiao.jianye.base.mqtt.bean.DlcMqttConnectOptions;
import com.dht.xiao.jianye.base.mqtt.callback.BaseMqttCallBack;
import com.dht.xiao.jianye.base.mqtt.log.MqttLogger;
import com.dht.xiao.jianye.base.mqtt.log.MqttLoggerFactory;
import com.dht.xiao.jianye.base.util.GsonUtil;
import io.reactivex.Observable;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.Consumer;
import io.reactivex.schedulers.Schedulers;

/**
 * @author :      fangbingran
 * @aescription : mqtt管理器
 * @date :        2019/06/13  17:56
 */
public class MqttMgr {
    private static class InstanceMqttMgr {
        private static final MqttMgr INSTANCE = new MqttMgr();
    }

    public static MqttMgr get() {
        return MqttMgr.InstanceMqttMgr.INSTANCE;
    }

    /**
     * mqtt客户端
     */
    private MqttAndroidClient mMqttAndroidClient;
    /**
     * mqtt配置
     */
    private MqttConnectOptions mMqttConnectOptions;
    /**
     * 回调
     */
    private BaseMqttCallBack mBaseMqttCallBack;
    /**
     * 信息配置
     */
    private DlcMqttConnectOptions mDlcMqttConnectOptions;
    /**
     * 全部的topics
     */
    private final ArrayMap<String, Integer> allTopics = new ArrayMap<>();

    /**
     * 收到的消息
     */
    private ArrayMap<String, Long> mMqttMessageData = new ArrayMap<>();
    /**
     * 去重的过时时间,默认半个小时
     */
    private int mDuplicateOutTimeSeconds = 30 * 60 * 1000;
    /**
     * 是否取消,默认为true
     */
    private boolean canDoConnect = true;
    /**
     * 是否去重,默认为false
     */
    private boolean isDuplicate = false;
    private Disposable mDisposable;
    private Context context;

    public void init(Context context, DlcMqttConnectOptions mqttInfoBean, BaseMqttCallBack baseMqttCallBack) {
        init(context, mqttInfoBean, null, baseMqttCallBack);
    }

    public void init(Context context, DlcMqttConnectOptions mqttInfoBean, MqttLogger mqttLogger, BaseMqttCallBack baseMqttCallBack) {
        if (mqttInfoBean == null) {
            throw new ApiException(ApiExceptionCode.MQTT_NOT_INIT_ERROR, "mqttInfoBean不能为空值");
        }
        if (context == null) {
            throw new ApiException(ApiExceptionCode.MQTT_NOT_INIT_ERROR, "context不能为空值");
        }
        this.context = context;
        mBaseMqttCallBack = baseMqttCallBack;
        MqttLoggerFactory.createLogger(mqttLogger);
        mDlcMqttConnectOptions = mqttInfoBean;
        mMqttAndroidClient = new MqttAndroidClient(context.getApplicationContext(),
                mqttInfoBean.getServerUri(), mqttInfoBean.getClientIdCurrentTimeMillis());
        mMqttAndroidClient.setCallback(mqttCallbackExtended);
        mMqttConnectOptions = new MqttConnectOptions();
        // 清除缓存
        mMqttConnectOptions.setCleanSession(mDlcMqttConnectOptions.isCleanSession());
        // 设置超时时间，单位：秒
        mMqttConnectOptions.setConnectionTimeout(mDlcMqttConnectOptions.getTimeOutSeconds());
        // 心跳包发送间隔，单位：秒
        mMqttConnectOptions.setKeepAliveInterval(mDlcMqttConnectOptions.getKeepAliveIntervalSeconds());
        // 用户名
        mMqttConnectOptions.setUserName(mDlcMqttConnectOptions.getUserName());
        // 密码
        mMqttConnectOptions.setPassword(mDlcMqttConnectOptions.getPassWord().toCharArray());
        connect(5);
    }

    public void setMqttLogger(MqttLogger mqttLogger) {
        MqttLoggerFactory.createLogger(mqttLogger);
    }

    public void setDuplicateOutTimeSeconds(int timeSeconds) {
        mDuplicateOutTimeSeconds = timeSeconds * 1000;
    }

    private void connect(long delay) {
        disposable();
        mDisposable = Observable.timer(delay, TimeUnit.SECONDS).subscribeOn(Schedulers.newThread()).subscribe(new Consumer<Long>() {
            @Override
            public void accept(Long aLong) throws Exception {
                MqttLoggerFactory.info("Observable-->connect)()");
                // 开始连接
                connect();
            }
        });
    }

    /**
     * 是否去重
     *
     * @param isDuplicate
     */
    public void setDuplicate(boolean isDuplicate) {
        this.isDuplicate = isDuplicate;
    }

    /**
     * 连接MQTT服务器
     */
    public void connect() {
        if (mMqttAndroidClient == null || mDlcMqttConnectOptions == null) {
            throw new ApiException(ApiExceptionCode.MQTT_NOT_INIT_ERROR, "请初始化init()");
        }
        MqttLoggerFactory.info("connect-->canDoConnect:" + canDoConnect + "isConnected:" + isConnected());
        if (canDoConnect || !isConnected()) {
            try {
                MqttLoggerFactory.info("connect-->开始连接");
                mMqttAndroidClient.connect(mMqttConnectOptions, context.getApplicationContext(), iMqttConnectListener);
                canDoConnect = false;
            } catch (MqttException e) {
                e.printStackTrace();
                MqttLoggerFactory.error(e.toString());
            }
        }
    }

    /**
     * 连接MQTT服务器
     */
    public void close() {
        try {
            mMqttAndroidClient.close();
        } catch (Exception e) {
            MqttLoggerFactory.error(e.toString());
        }
    }

    /**
     * 清空客户端
     */
    public void cleanClient() {
        try {
            MqttLoggerFactory.error("cleanClient");
            mMqttAndroidClient.close();
            mMqttAndroidClient.disconnect();
            mMqttAndroidClient.unregisterResources();
            mMqttAndroidClient = null;
        } catch (Exception e) {
            MqttLoggerFactory.error(e.toString());
        }
    }

    private void disposable() {
        if (mDisposable != null && !mDisposable.isDisposed()) {
            mDisposable.dispose();
        }
    }

    /**
     * 自动订阅主题
     *
     * @param topics 主题
     * @param qos    策略
     */
    public void subscribe(String[] topics, int[] qos) {

        if (topics.length != qos.length) {
            throw new ApiException(ApiExceptionCode.MQTT_ERROR, "topics和qos的数组大小必须相同");
        }
        for (int i = 0; i < topics.length; i++) {
            subscribe(topics[i], qos[i], true);
        }

    }

    /**
     * 自动订阅主题
     *
     * @param topic 主题
     * @param qos   策略
     */
    public void subscribe(final String topic, final int qos) {
        subscribe(topic, qos, true);
    }

    private void allSubscribe() {
        for (int i = 0; i < allTopics.size(); i++) {
            String key = allTopics.keyAt(i);
            int value = allTopics.valueAt(i);
            subscribe(key, value, false);
        }
    }

    /**
     * 自动订阅主题
     *
     * @param topic 主题
     * @param qos   策略
     */
    private void subscribe(final String topic, final int qos, boolean isAddAllTops) {
        try {
            if (isAddAllTops) {
                synchronized (allTopics) {
                    allTopics.put(topic, qos);
                }
            }
            if (isConnected()) {
                // 订阅topic话题
                MqttLoggerFactory.info("execute subscribe -- qos = " + qos);
                mMqttAndroidClient.subscribe(topic, qos);
            } else {
                MqttLoggerFactory.info("execute subscribe isNotConnected");
            }
        } catch (Exception e) {
            MqttLoggerFactory.error(e.toString());
        }
    }

    /**
     * 私自订阅主题
     *
     * @param topic 主题
     * @param qos   策略
     */
    public void subscribe(String topic, int qos, IMqttActionListener iMqttActionListener, IMqttMessageListener iMqttMessageListener) {
        try {
            // 订阅topic话题
            if (isConnected()) {
                MqttLoggerFactory.info("execute subscribe -- qos = " + qos);
                mMqttAndroidClient.subscribe(topic, qos, null, iMqttActionListener, iMqttMessageListener);
            } else {
                MqttLoggerFactory.info("execute subscribe isNotConnected");
            }
        } catch (Exception e) {
            MqttLoggerFactory.error(e.toString());
        }
    }

    /**
     * 取消订阅主题
     *
     * @param topic 主题
     */
    public void unsubscribe(String topic) {
        try {
            // 取消订阅主题
            MqttLoggerFactory.info("execute unsubscribe --  " + topic);
            mMqttAndroidClient.unsubscribe(topic);
        } catch (Exception e) {
            MqttLoggerFactory.error(e.toString());
        }
    }

    /**
     * 取消订阅主题
     *
     * @param topic 主题
     */
    public void unsubscribe(String[] topic) {
        try {
            // 取消订阅主题
            MqttLoggerFactory.info("execute unsubscribe --  " + topic.toString());
            mMqttAndroidClient.unsubscribe(topic);
        } catch (Exception e) {
            MqttLoggerFactory.error(e.toString());
        }
    }

    /**
     * 断开连接
     */
    public void disconnect() {
        try {
            mMqttAndroidClient.disconnect();
        } catch (Exception e) {
            MqttLoggerFactory.error(e.toString());
        }
    }

    /**
     * 判断连接是否连接
     */
    public boolean isConnected() {
        try {
            return mMqttAndroidClient.isConnected();
        } catch (Exception e) {
            MqttLoggerFactory.error(toString());
        }
        return false;
    }


    /**
     * 发送订阅主题
     */
    public void publish(String topic, int qos, byte[] payload, boolean retained) {
        publish(topic, qos, payload, retained, null);
    }

    /**
     * 发送订阅主题
     */
    public void publish(String topic, int qos, byte[] payload) {
        publish(topic, qos, payload, false);
    }

    /**
     * 发送订阅主题
     */
    public void publish(String topic, int qos, Object object) {
        byte[] payload = GsonUtil.getInstance().parseObjToJsonStr(object).getBytes();
        publish(topic, qos, payload, false);
    }

    /**
     * 发送订阅主题
     */
    public void publish(String topic, int qos, byte[] payload, boolean retained, IMqttActionListener iMqttActionListener) {
        IMqttToken iMqttToken = null;
        try {
            MqttLoggerFactory.info("execute publish --  " + topic.toString() + "--消息:" + new String(payload));
            iMqttToken = mMqttAndroidClient.publish(topic, payload, qos, retained, null, iMqttActionListener);
        } catch (Exception e) {
            if (iMqttActionListener != null) {
                iMqttActionListener.onFailure(iMqttToken, e.getCause());
            }
            MqttLoggerFactory.error("publish-->>Exception:" + e.toString());
        }
    }


    /**
     * MQTT监听是否连接成功
     */
    private IMqttActionListener iMqttConnectListener = new IMqttActionListener() {


        @Override
        public void onSuccess(IMqttToken asyncActionToken) {
            MqttLoggerFactory.info("mqtt connect success ");
            allSubscribe();
            if (mBaseMqttCallBack != null) {
                mBaseMqttCallBack.onSuccess(asyncActionToken);
            }
        }

        @Override
        public void onFailure(IMqttToken asyncActionToken, Throwable exception) {
            MqttLoggerFactory.info("mqtt connect failed ");
            mqttFailure(exception);
        }
    };

    private void mqttFailure(Throwable cause) {
        if (mDlcMqttConnectOptions != null && mDlcMqttConnectOptions.isAutoReconnect()) {
            MqttLoggerFactory.info("mqtt reConnect ");
            setClientId();
            canDoConnect = true;
            connect(mDlcMqttConnectOptions.getAutoReconnectTimeSeconds());

        }
        if (mBaseMqttCallBack != null) {
            mBaseMqttCallBack.onFailure(cause);
        }
    }

    private void setClientId() {
        if (mDlcMqttConnectOptions.isCleanClient()) {
            cleanClient();
            mDlcMqttConnectOptions.refreshClientIdCurrentTimeMillis();
            mMqttAndroidClient = new MqttAndroidClient(context.getApplicationContext(),
                    mDlcMqttConnectOptions.getServerUri(), mDlcMqttConnectOptions.getClientIdCurrentTimeMillis());
            mMqttAndroidClient.setCallback(mqttCallbackExtended);
        }
    }

    /**
     * MQTT监听并且接受消息
     */
    private MqttCallbackExtended mqttCallbackExtended = new MqttCallbackExtended() {


        @Override
        public void connectComplete(boolean reconnect, String serverURI) {
            MqttLoggerFactory.info("mqtt connectComplete ");
        }

        @Override
        public void connectionLost(Throwable cause) {
            MqttLoggerFactory.info("mqtt connectionLost ");
            mqttFailure(cause);
        }

        @Override
        public void messageArrived(String topic, MqttMessage message) throws Exception {
            String msgContent = new String(message.getPayload());
            boolean isHasMessage = isHasMessage(message);
            if (isHasMessage) {
                String detailLog = topic + ";qos:" + message.getQos() + ";retained:" + message.isRetained() + ";msgContent:" + msgContent + ";id:" + message.getId();
                MqttLoggerFactory.info("重复消息,messageArrived:" + detailLog);
            } else {
                String detailLog = topic + ";qos:" + message.getQos() + ";retained:" + message.isRetained() + ";msgContent:" + msgContent + ";id:" + message.getId();
                MqttLoggerFactory.info("messageArrived:" + detailLog);
                if (mBaseMqttCallBack != null) {
                    mBaseMqttCallBack.messageArrived(topic, msgContent, message.getId());
                }
            }
        }

        @Override
        public void deliveryComplete(IMqttDeliveryToken token) {
            MqttLoggerFactory.info("mqtt deliveryComplete ");
        }
    };

    private boolean isHasMessage(MqttMessage message) {
        if (!isDuplicate) {
            //如果不设置去重则直接回调,默认去重
            return false;
        }
        synchronized (mMqttMessageData) {
            boolean isHasMessage = false;
            for (Iterator<Map.Entry<String, Long>> it = mMqttMessageData.entrySet().iterator(); it.hasNext(); ) {
                Map.Entry<String, Long> item = it.next();
                long value = item.getValue();
                if (value > 0 && System.currentTimeMillis() - value >= mDuplicateOutTimeSeconds) {
                    it.remove();
                    continue;
                }
                if (TextUtils.equals(item.getKey(), message.getId() + "")) {
                    isHasMessage = true;
                }
            }
            if (!isHasMessage) {
                mMqttMessageData.put(message.getId() + "", System.currentTimeMillis());
            }
            return isHasMessage;
        }

    }
}

